package com.atguigu.userprofil;

import com.atguigu.userprofile.bean.TagInfo;
import com.atguigu.userprofile.dao.TagInfoDAO;
import com.atguigu.userprofile.dao.TaskInfoDAO;
import com.atguigu.userprofile.util.MyClickhouseUtil;
import com.atguigu.userprofile.util.MyPropertiesUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;

import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;

public class TaskOutClickHouseApp {
    public static void main(String[] args) {

        SparkConf sparkConf = new SparkConf().setAppName("TaskOutClickHouseApp");//.setMaster("local[*]");
        SparkSession sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate();

        //获取传进来的第二个参数（业务日期）
        String busiDate = args[1];

        Properties properties = MyPropertiesUtil.load("config.properties");
        //hive的库名
        String hiveDbName = properties.getProperty("user-profile.dbname");
        //clickhouse库名
        String ckDbName = properties.getProperty("clickhouse.dbname");
        //Clickhouse Url
        String ckUrl = properties.getProperty("clickhouse.url");

        //表名
        String tableName = "up_tag_merge_" + busiDate.replace("-", "");

        //1.在Clickhouse中建表
        //表名？ 字段？ 引擎？ order by？
        /**
         *  create table if not exists user_profile.up_tag_merge_20200614
         *   (uid String, tg_person_base_gender String,tg_person_base_agegroup String)
         *   engine=MergeTree
         *   order by uid
         */

        //1.2获取启用的标签code
        List<TagInfo> tagInfoList = TagInfoDAO.getTagInfoListWithOn();
        List<String> codeStringList = tagInfoList.stream().map(tagInfo -> tagInfo.getTagCode().toLowerCase() + " String").collect(Collectors.toList());
        String tagCodeSQL = StringUtils.join(codeStringList, ",");

        //删除表的语句
        String dropTableSQL = "drop table if exists " + ckDbName + "." + tableName;
        MyClickhouseUtil.executeSql(dropTableSQL);

        //建表语句
        String createTableSQL = "create table if not exists "+ckDbName+"."+tableName+" \n" +
                "         (uid String,"+tagCodeSQL+") \n" +
                "         engine=MergeTree \n" +
                "       order by uid";

        System.out.println(createTableSQL);

        //怎么去执行这个建表语句让其在Clickhouse中建表  答：通过jdbc的方式执行sql去Clickhouse建表
        MyClickhouseUtil.executeSql(createTableSQL);

        //2.查询出hive中的数据
        Dataset<Row> sql = sparkSession.sql("select * from " + hiveDbName + "." + tableName);

        //保存到哪?  答：Dataset

        //3.将查询出的数据写入Clickhouse
        Properties ckProperties = new Properties();
        sql.write().mode(SaveMode.Append)
                .option("driver","ru.yandex.clickhouse.ClickHouseDriver")
                .option("batchsize",500) //批量提交1.减少连接 网络IO次数 2.减少磁盘碎片
                .option("isolationLevel","NONE")   //事务关闭
                .option("numPartitions", "4") // 设置并发
                .jdbc(ckUrl, tableName, ckProperties);
    }
}
